package com.bdi.mqtt_broker.mqtt_protocol;

import com.bdi.mqtt_broker.constant.Constants;
import com.bdi.mqtt_broker.model.SubscribeStore;
import com.bdi.mqtt_broker.service.ISubscribeStoreService;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * @author WangNing yogehaoren@gmial.com <br>
 * @since 1.0
 */
@Slf4j
public class Subscribe {

    private final ISubscribeStoreService subscribeStoreService;

    public Subscribe(ISubscribeStoreService subscribeStoreService){
        this.subscribeStoreService = subscribeStoreService;
    }

    public void processSubscribe(Channel channel, MqttSubscribeMessage msg) {

        List<MqttTopicSubscription> topicSubscriptions = msg.payload().topicSubscriptions();
        if(validTopicFilter(topicSubscriptions)){

            String clientId = (String) channel.attr(AttributeKey.valueOf(Constants.CHANNEL_ATTRIBUTE_CLIENT_ID)).get();
            List<Integer> mqttQoSList = new ArrayList<Integer>();
            topicSubscriptions.forEach(topicSubscription -> {
                String topicFilter = topicSubscription.topicName();
                MqttQoS mqttQoS = topicSubscription.qualityOfService();
                SubscribeStore subscribeStore = new SubscribeStore(clientId, topicFilter, mqttQoS.value());
                subscribeStoreService.put(topicFilter, subscribeStore);
                mqttQoSList.add(mqttQoS.value());
                log.debug("SUBSCRIBE - clientId: {}, topFilter: {}, QoS: {}", clientId, topicFilter, mqttQoS.value());
            });
            MqttSubAckMessage subAckMessage = (MqttSubAckMessage) MqttMessageFactory.newMessage(
                    new MqttFixedHeader(MqttMessageType.SUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
                    MqttMessageIdVariableHeader.from(msg.variableHeader().messageId()),
                    new MqttSubAckPayload(mqttQoSList));
            channel.writeAndFlush(subAckMessage);

        }

    }


    private boolean validTopicFilter(List<MqttTopicSubscription> topicSubscriptions) {
        for (MqttTopicSubscription topicSubscription : topicSubscriptions) {
            String topicFilter = topicSubscription.topicName();
            if(!StringUtils.hasText(topicFilter)) {return false;}
            // 不以/开头的,以#或+符号开头的、以/符号结尾的订阅按非法订阅处理, 这里没有参考标准协议
            if ( !topicFilter.startsWith("/")||topicFilter.startsWith("+")||topicFilter.startsWith("#") || topicFilter.endsWith( "/"))
                return false;
            if (topicFilter.endsWith( "#")) {
                // 如果出现多个#符号的订阅按非法订阅处理
                if(StringUtils.countOccurrencesOf(topicFilter, "#")>2){
                    return false;
                }
            }
            if (topicFilter.endsWith("+")) {
                //如果出现多个+符号的订阅按非法订阅处理
                if(StringUtils.countOccurrencesOf(topicFilter, "+")>2){
                    return false;
                }
            }
        }
        return true;
    }

}
